package com.fanxuankai.ksql.function.udaf.count;

import com.google.common.collect.Lists;
import io.confluent.ksql.function.udaf.TableUdaf;
import io.confluent.ksql.function.udaf.UdafDescription;
import io.confluent.ksql.function.udaf.UdafFactory;

import java.nio.ByteBuffer;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.List;

/**
 * @author fanxuankai
 */
@UdafDescription(name = "COUNT_DISTINCT_PRO", description = "去重计数", author = "fanxuankai")
public final class CountDistinct {

    private CountDistinct() {
    }

    @UdafFactory(description = "Integer 类型")
    public static TableUdaf<Integer, List<Integer>, Long> countDistinctInteger() {
        return new Function<>();
    }

    @UdafFactory(description = "Long 类型")
    public static TableUdaf<Long, List<Long>, Long> countDistinctLong() {
        return new Function<>();
    }

    @UdafFactory(description = "Double 类型")
    public static TableUdaf<Double, List<Double>, Long> countDistinctDouble() {
        return new Function<>();
    }

    @UdafFactory(description = "String 类型")
    public static TableUdaf<String, List<String>, Long> countDistinctString() {
        return new Function<>();
    }

    @UdafFactory(description = "Boolean 类型")
    public static TableUdaf<Boolean, List<Boolean>, Long> countDistinctBoolean() {
        return new Function<>();
    }

    @UdafFactory(description = "Timestamp 类型")
    public static TableUdaf<Timestamp, List<Timestamp>, Long> countDistinctTimestamp() {
        return new Function<>();
    }

    @UdafFactory(description = "Time 类型")
    public static TableUdaf<Time, List<Time>, Long> countDistinctTime() {
        return new Function<>();
    }

    @UdafFactory(description = "Date 类型")
    public static TableUdaf<Date, List<Date>, Long> countDistinctDate() {
        return new Function<>();
    }

    @UdafFactory(description = "ByteBuffer 类型")
    public static TableUdaf<ByteBuffer, List<ByteBuffer>, Long> countDistinctByteBuffer() {
        return new Function<>();
    }

    private static final class Function<T> implements TableUdaf<T, List<T>, Long> {

        @Override
        public List<T> initialize() {
            return Lists.newArrayList();
        }

        @Override
        public List<T> aggregate(final T thisValue, final List<T> aggregate) {
            if (thisValue != null && !aggregate.contains(thisValue)) {
                aggregate.add(thisValue);
            }
            return aggregate;
        }

        @Override
        public List<T> merge(final List<T> aggOne, final List<T> aggTwo) {
            for (final T thisEntry : aggTwo) {
                if (!aggOne.contains(thisEntry)) {
                    aggOne.add(thisEntry);
                }
            }
            return aggOne;
        }

        @Override
        public Long map(final List<T> agg) {
            return (long) agg.size();
        }

        @Override
        public List<T> undo(T valueToUndo, List<T> aggregateValue) {
            if (valueToUndo != null) {
                aggregateValue.remove(valueToUndo);
            }
            return aggregateValue;
        }
    }

}
